/*
package com.wsoft.mq.test.confirmMessage;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.wsoft.utils.RabbitMQUtil;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

*/
/**
 * 发布确认模式
 *  三种模式: 1. 单个确认
 *           2. 批量确认
 *           3. 异步批量确认
 * @Author z001
 * @Date 2025/1/9 10:59
 *//*

public class ConfirmMessage {

    //发布消息的个数
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception{
        //1.单个确认模式
//        publicMessageIndividually(); //发布1000个确认消息,耗时20143ms

        //2.批量发布确认
//        publicMessageBatch();//发布1000个确认消息,耗时282ms

        //3.异步发布确认
        publicMessageAsync();//发布1000个确认消息,耗时45ms

    }



    */
/**
     * 异步发布确认
     * @throws Exception
     *//*

    public static   void  publicMessageAsync() throws  Exception{
        Channel channel = RabbitMQUtil.getChannel();
        String queue = UUID.randomUUID().toString();
        channel.queueDeclare(queue,false,false,true,null);
        //开启发布确认模式
        channel.confirmSelect();
        long begin = System.currentTimeMillis();

        */
/**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         *//*

        ConcurrentSkipListMap<Long, String> outstandingConfirms = new
                ConcurrentSkipListMap<>();


        */
/**
         * 确认收到消息的一个回调
         * 1.消息序列号
         * 2.true 可以确认小于等于当前序列号的消息
         * false 确认当前序列号消息
         *//*

        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {//是否批量 是批量就批量删除
                //返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed =
                        outstandingConfirms.headMap(deliveryTag, true);
                //清除该部分未确认消息
                confirmed.clear();//删除已确认的消息，剩下的就是未确认的消息
            }else{//不是批量就删除单个， (异步一般都是单个确认)
                //只清除当前序列号的消息
                outstandingConfirms.remove(deliveryTag);
            }
        };
        */
/**
         * 消息确认失败 回调
         *//*

        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("发布的消息"+message+"未被确认，序列号"+deliveryTag);
        };

        //添加一个异步消息监听器，监听哪些消息成功了 哪些消息失败了
        channel.addConfirmListener(ackCallback,nackCallback);


        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            */
/**
             * channel.getNextPublishSeqNo()获取下一个消息的序列号
             * 通过序列号与消息体进行一个关联
             * 全部都是未确认的消息体
             *//*

            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", queue, null, message.getBytes());
        }

        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个确认消息,耗时" + (end - begin) +
                "ms");
    }


    */
/**
     * 批量发布确认
     * @throws Exception
     *//*

    public static   void  publicMessageBatch() throws  Exception{
        Channel channel = RabbitMQUtil.getChannel();
        String queue = UUID.randomUUID().toString();
        channel.queueDeclare(queue,false,false,true,null);

        //开启发布确认模式
        channel.confirmSelect();
        long begin = System.currentTimeMillis();

        //批量确认消息大小
        int batchSize = 100;
        //未确认消息个数
        int outstandingMessageCount = 0;

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queue,null,message.getBytes());
            outstandingMessageCount++;
            if (outstandingMessageCount == batchSize) {
                channel.waitForConfirms();
                outstandingMessageCount = 0;
            }

        }

        //为了确保还有剩余没有确认消息 再次确认
        if (outstandingMessageCount > 0) {
            channel.waitForConfirms();
        }

        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个确认消息,耗时" + (end - begin) +
                "ms");
    }


    */
/**
     * 单个确认 发一条 - 确认一条
     * @throws Exception
     *//*

    public static  void  publicMessageIndividually() throws  Exception{
        Channel channel = RabbitMQUtil.getChannel();
        String queue = UUID.randomUUID().toString();
        channel.queueDeclare(queue,false,false,true,null);

        //开启发布确认模式
        channel.confirmSelect();
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queue,null,message.getBytes());

            //等待服务器返回，若服务器端返回false或者超时，则生产者可以重发消息
            boolean b = channel.waitForConfirms();
            if(b){
                System.out.println("消息发送成功");
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个确认消息,耗时" + (end - begin) +
                "ms");

    }




}
*/
